package mqs

import (
	"context"
	"gitee.com/lailonghui/vehicle-supervision-framework/pkg/asserts"
	"gitee.com/lailonghui/vehicle-supervision-framework/pkg/loggers"
	"github.com/go-kit/kit/endpoint"
	amqptransport "github.com/go-kit/kit/transport/amqp"
	"github.com/streadway/amqp"
	"go.uber.org/zap"
)

type ConsumerConfig struct {
	// --- 以下是qos配置
	PrefetchCount int

	PrefetchSize int

	Global bool
	// --- 以下是consumer配置
	Queue string

	ConsumerName string

	Exclusive bool

	NoLocal bool

	NoWait bool

	Args amqp.Table

	// --- 以下配置
	// go kit
	Handler endpoint.Endpoint
	// 入参解析
	DecodeRequestFunc amqptransport.DecodeRequestFunc
	// 出参渲染
	EncodeResponseFunc amqptransport.EncodeResponseFunc
	// 返回消息发布到新的队列
	ResponsePublisher amqptransport.ResponsePublisher
}

// 新建Config
func NewConfig(queue, consumerName string, Handler endpoint.Endpoint, DecodeRequestFunc amqptransport.DecodeRequestFunc, EncodeResponseFunc amqptransport.EncodeResponseFunc, ResponsePublisher amqptransport.ResponsePublisher) ConsumerConfig {
	return ConsumerConfig{
		PrefetchCount:      1,
		PrefetchSize:       0,
		Global:             false,
		Queue:              queue,
		ConsumerName:       consumerName,
		Exclusive:          false,
		NoLocal:            false,
		NoWait:             false,
		Args:               nil,
		Handler:            Handler,
		DecodeRequestFunc:  DecodeRequestFunc,
		EncodeResponseFunc: EncodeResponseFunc,
		ResponsePublisher:  ResponsePublisher,
	}
}

func NewCommonConfig(queue, consumerName string, Handler endpoint.Endpoint, DecodeRequestFunc amqptransport.DecodeRequestFunc) ConsumerConfig {
	return NewConfig(queue, consumerName, Handler, DecodeRequestFunc, EncodeJSONResponse, amqptransport.NopResponsePublisher)
}

// 新建消费者
func NewConsumer(connection *amqp.Connection, config ConsumerConfig) {
	channel, err := connection.Channel()
	asserts.Nil(err, err)

	// 设置请求参数，比如每次拉取多少数据
	err = channel.Qos(
		config.PrefetchCount, // prefetch count
		config.PrefetchSize,  // prefetch size
		config.Global,        // global
	)
	asserts.Nil(err, err)

	// 请求消费数据
	msgs, err := channel.Consume(
		config.Queue,        // queue
		config.ConsumerName, // consumer
		false,               // auto-ack
		config.Exclusive,    // exclusive
		config.NoLocal,      // no-local
		config.NoWait,       // no-wait
		config.Args,         // args
	)
	asserts.Nil(err, err)

	subscriber := amqptransport.NewSubscriber(config.Handler,
		config.DecodeRequestFunc,
		config.EncodeResponseFunc,
		amqptransport.SubscriberErrorEncoder(AckAndRepublish5TimesErrorEncoder),
		amqptransport.SubscriberBefore(LoggerDelivery),
		amqptransport.SubscriberAfter(amqptransport.SetAckAfterEndpoint(false)),
		amqptransport.SubscriberResponsePublisher(config.ResponsePublisher))

	serveDelivery := subscriber.ServeDelivery(channel)
	go func() {
		for d := range msgs {
			serveDelivery(&d)
		}
	}()
}

func LoggerDelivery(ctx context.Context, publishing *amqp.Publishing, delivery *amqp.Delivery) context.Context {
	loggers.Info("接收到消息", ctx, zap.String("MessageId", delivery.MessageId), zap.String("exchange", delivery.Exchange), zap.Any("Headers", delivery.Headers), zap.Uint64("deliveryTag", delivery.DeliveryTag))

	return ctx
}

// ack，并且重新发送到队列5次
func AckAndRepublish5TimesErrorEncoder(ctx context.Context, err error, deliv *amqp.Delivery, ch amqptransport.Channel, pub *amqp.Publishing) {

	loggers.Error("消费消息异常", ctx, zap.String("MessageId", deliv.MessageId), zap.String("exchange", deliv.Exchange), zap.Error(err))
	verrorCount := deliv.Headers["error_count"]
	var errcount int32 = 0
	if verrorCount != nil {
		errcount, _ = verrorCount.(int32)
	}
	errcount++
	if errcount >= 5 {
		_ = deliv.Nack(false, false)
	} else {
		_ = deliv.Ack(false)
		_ = ch.Publish(deliv.Exchange, deliv.RoutingKey, false, false, amqp.Publishing{
			Headers: map[string]interface{}{
				"error_count": errcount,
				"x-delay":     getDelayByErrorCount(errcount),
			},
			DeliveryMode: deliv.DeliveryMode,
			Priority:     deliv.Priority,
			Expiration:   deliv.Expiration,
			MessageId:    deliv.MessageId,
			Timestamp:    deliv.Timestamp,
			Type:         deliv.Type,
			UserId:       deliv.UserId,
			AppId:        deliv.AppId,
			Body:         deliv.Body,
		})
	}

}

// 通过错误次数，获取延迟时间
func getDelayByErrorCount(errcount int32) int32 {
	switch errcount {
	case 1:
		return 1000
	case 2:
		return 5000
	case 3:
		return 10000
	case 4:
		return 20000
	case 5:
		return 30000
	default:
		return 30000
	}
}
